package main.java.black;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * TopIpBolt
 *
 * @author zhangyimin
 * @version 1.0
 * @date 2018-11-13 下午2:54
 */
public class BlackListSplitBolt extends BaseRichBolt {

    private OutputCollector  collector;


    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
            this.collector=outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
//        1,201.105.101.102,http://mystore.jsp/?productid=1,2017020029,2,1
        String log=tuple.getString(0);

        String[] fields=log.split(",");
        if (fields.length==6){
            this.collector.emit(new Values(fields[0],new Long(1)));
        }
        this.collector.ack(tuple);

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("user_id","count"));
    }
}
